package com.whty.cms.filewps.util;

import com.whty.cms.filewps.common.model.EsAttachment;
import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.metrics.valuecount.ParsedValueCount;
import org.elasticsearch.search.aggregations.metrics.valuecount.ValueCountAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class EsUtil {
	
	private static String schema = "http";
	private static RestHighLevelClient restHighLevelClient;

    private static RestHighLevelClient restHighLevelClient1;
	
	private static final Logger logger = LogManager.getLogger(EsUtil.class);


    /**
     * 初始化es客户端
     * @return
     */
    public static void initEsClient() {
    	try {
    		final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                    //es账号密码
                    new UsernamePasswordCredentials(MemcacheUtil.getDatabaseObj("es-username"), MemcacheUtil.getDatabaseObj("es-password")));
    		// 初始化 RestClient, hostName 和 port 填写集群的内网 VIP 地址与端口
    		RestClientBuilder builder = RestClient.builder(new HttpHost("10.0.200.39", 30104, schema)
    				).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                            return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                    });
    		// 设置超时时间
    		builder.setMaxRetryTimeoutMillis(10000);
    		// 由Low Level Client构造High Level Client
    		restHighLevelClient = new RestHighLevelClient(builder);
    	}catch(Exception e) {
    		System.out.println(e.getMessage());
    		logger.error(e.getMessage());
    		e.printStackTrace();
    	}
    }

    //初始化平台ES
    public static void initEsClient1() {
        try {
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                    //es账号密码
                    new UsernamePasswordCredentials("elastic", "E1UgWpebBJNn"));
            // 初始化 RestClient, hostName 和 port 填写集群的内网 VIP 地址与端口
            RestClientBuilder builder = RestClient.builder(new HttpHost("10.0.200.39", Integer.parseInt("5060"), schema)
            ).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                    return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
            });
            // 设置超时时间
            builder.setMaxRetryTimeoutMillis(10000);
            // 由Low Level Client构造High Level Client
            restHighLevelClient1 = new RestHighLevelClient(builder);
        }catch(Exception e) {
            System.out.println(e.getMessage());
            logger.error(e.getMessage());
            e.printStackTrace();
        }
    }

    public static RestHighLevelClient getHighLevelClient1(){
        if(restHighLevelClient1==null){
            initEsClient1();
        }
        return restHighLevelClient1;
    }
    
    public static RestHighLevelClient getHighLevelClient(){
        if(restHighLevelClient==null){
        	initEsClient();
        }
        return restHighLevelClient;
    }

    public static void close() {
        if (restHighLevelClient != null) {
            try {
                restHighLevelClient.close();
            } catch (Exception e) {
            	logger.error("close rest client exception:"+ e.toString());
            }
        }
    }

    public static void deleDoc(List<String> fids) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        for(String fid: fids) {
            DeleteRequest deleteRequest = new DeleteRequest(MemcacheUtil.getDatabaseObj("es-index"),MemcacheUtil.getDatabaseObj("es-type"),fid);
            bulkRequest.add(deleteRequest);
        }
        getHighLevelClient().bulk(bulkRequest,RequestOptions.DEFAULT);
    }

    /**
     * 插入或修改数据
     */
    public static void addOrUpdateDoc(String id, String source) {
    	IndexRequest indexRequest = new IndexRequest(MemcacheUtil.getDatabaseObj("es-index"), MemcacheUtil.getDatabaseObj("es-type"), id).source(source, XContentType.JSON);
    	try {
            // 获取响应结果
            IndexResponse indexResponse = getHighLevelClient().index(indexRequest,RequestOptions.DEFAULT);
            String index = indexResponse.getIndex();
            String type = indexResponse.getType();
            String ids = indexResponse.getId();
            long version = indexResponse.getVersion();

            if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                //logger.info("doc indexed, index: "+ index +", type:"+ type +",id:"+ ids+",version:"+version);
             } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                //logger.info("doc updated, index: "+ index +", type:"+ type +",id:"+ ids+",version:"+version);
            }
        }catch(ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                logger.error("version conflict");
            }
        }catch(Exception e){
            logger.error("execute index api failed, "+ e.toString());
        }     
    }


    public static void addOrUpdateData(String id, String source) {
        IndexRequest indexRequest = new IndexRequest("tyedu_fct_resource_info_daily_total", "data", id).source(source, XContentType.JSON);
        try {
            // 获取响应结果
            IndexResponse indexResponse = getHighLevelClient1().index(indexRequest,RequestOptions.DEFAULT);
            String index = indexResponse.getIndex();
            String type = indexResponse.getType();
            String ids = indexResponse.getId();
            long version = indexResponse.getVersion();

            if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                //logger.info("doc indexed, index: "+ index +", type:"+ type +",id:"+ ids+",version:"+version);
            } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                //logger.info("doc updated, index: "+ index +", type:"+ type +",id:"+ ids+",version:"+version);
            }
        }catch(ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                logger.error("version conflict");
            }
        }catch(Exception e){
            logger.error("execute index api failed, "+ e.toString());
        }
    }

    /**
     * 根据条件查询Count
     * @param esAttachment
     * @return
     */
    public static Long searchCountDoc(EsAttachment esAttachment) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        String fileMd5 = esAttachment.getFileMd5();
        if(StringUtils.isNotBlank(fileMd5)) {
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("fileMd5", fileMd5));
        }
        ValueCountAggregationBuilder valueCountAggregationBuilder = AggregationBuilders.count("fidSum").field("fid");
        sourceBuilder.aggregation(valueCountAggregationBuilder);
        sourceBuilder.query(boolQueryBuilder);

        SearchRequest searchRequest = new SearchRequest(MemcacheUtil.getDatabaseObj("es-index"));
        searchRequest.types(MemcacheUtil.getDatabaseObj("es-type"));
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            Aggregations aggregations = response.getAggregations();
            Map<String, Aggregation> map =  aggregations.asMap();
            ParsedValueCount parsedCount = (ParsedValueCount) map.get("fidSum");
            return parsedCount.getValue();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return 0L;
    }
    /**
     * 根据主键id查询信息
     * @param fid
     */
    public static JSONObject searchDoc(String fid) {
    	JSONObject json = new JSONObject();
    	
    	Object attach = MemcacheUtil.getObj(fid);
    	if(null!=attach) {
    		json = JSONObject.fromObject(attach);
    	}else {
    		// 查询文档
    		GetRequest getRequest = new GetRequest(MemcacheUtil.getDatabaseObj("es-index"), MemcacheUtil.getDatabaseObj("es-type"),fid);
    		try {
    			// 获取响应结果
    			GetResponse getResponse = getHighLevelClient().get(getRequest,RequestOptions.DEFAULT);
    			String index = getResponse.getIndex();
    			String type = getResponse.getType();
    			String id = getResponse.getId();
    			if (getResponse.isExists()) {
    				long version = getResponse.getVersion();
    				String sourceAsString = getResponse.getSourceAsString();
    				json = JSONObject.fromObject(sourceAsString);
    				//logger.info("get doc, index: "+ index +", type:"+ type +",id:"+ id+",version:"+version +", source:"+ sourceAsString);
    			}
    		}catch (ElasticsearchException e) {
    			if (e.status() == RestStatus.NOT_FOUND) {
    				logger.warn("doc not found");
    			}
    		}
    		catch(Exception e){
    			logger.error("execute get api failed, "+ e.toString());
    		}
    	}
        return json;
    }



    public static JSONObject searchData(String fid) {
        JSONObject json = new JSONObject();


            // 查询文档
            GetRequest getRequest = new GetRequest("tyedu_fct_resource_info_daily_total", "data",fid);
            try {
                // 获取响应结果
                GetResponse getResponse = getHighLevelClient1().get(getRequest,RequestOptions.DEFAULT);
                String index = getResponse.getIndex();
                String type = getResponse.getType();
                String id = getResponse.getId();
                if (getResponse.isExists()) {
                    long version = getResponse.getVersion();
                    String sourceAsString = getResponse.getSourceAsString();
                    json = JSONObject.fromObject(sourceAsString);
                    logger.info("get doc, index: "+ index +", type:"+ type +",id:"+ id+",version:"+version +", source:"+ sourceAsString);
                }
            }catch (ElasticsearchException e) {
                if (e.status() == RestStatus.NOT_FOUND) {
                    logger.warn("doc not found");
                }
            }
            catch(Exception e){
                logger.error("execute get api failed, "+ e.toString());
            }

        return json;
    }


    
    public static JSONObject criteriaQuery(SearchSourceBuilder sourceBuilder) {
    	JSONObject json = new JSONObject();
    	SearchRequest searchRequest = new SearchRequest(MemcacheUtil.getDatabaseObj("es-index"));
    	searchRequest.types(MemcacheUtil.getDatabaseObj("es-type"));
    	searchRequest.source(sourceBuilder);
    	SearchResponse response = null;
    	try {
    		List<String> sourceList = new ArrayList<>();
    		response = getHighLevelClient().search(searchRequest,RequestOptions.DEFAULT);
    		SearchHits hits= response.getHits();
            for (SearchHit searchHit : hits) {
                String sourceAsMap = searchHit.getSourceAsString();
                sourceList.add(sourceAsMap);
            }
            json.put("count", hits.totalHits);
            //System.out.println(sourceList.toString());
            json.put("data",sourceList);
            
    	}catch(Exception e) {
    		System.out.println(e.toString());
    		logger.error("criteriaQuery failed, "+ e.toString());
    	}
    	return json;
    }

    public static JSONObject criteriaQuery1(SearchSourceBuilder sourceBuilder) {
        JSONObject json = new JSONObject();
        SearchRequest searchRequest = new SearchRequest("tyedu_fct_resource_info_daily_total");
        searchRequest.types("data");
        searchRequest.source(sourceBuilder);
        SearchResponse response = null;
        try {
            List<Map> sourceList = new ArrayList<>();
            response = getHighLevelClient1().search(searchRequest,RequestOptions.DEFAULT);
            SearchHits hits= response.getHits();
            for (SearchHit searchHit : hits) {
                Map sourceAsMap = searchHit.getSourceAsMap();
                sourceAsMap.put("orginId", searchHit.getId());
                sourceList.add(sourceAsMap);
            }
            json.put("count", hits.totalHits);
            System.out.println(sourceList.toString());
            json.put("data",sourceList);

        }catch(Exception e) {
            System.out.println(e.toString());
            logger.error("criteriaQuery failed, "+ e.toString());
        }
        return json;
    }

    /**
     * 插入或修改数据
     */
    public static void addOrUpdatePart(String id, String source) {
    	IndexRequest indexRequest = new IndexRequest(MemcacheUtil.getDatabaseObj("es-index-part"), MemcacheUtil.getDatabaseObj("es-type-part"), id).source(source, XContentType.JSON);
    	try {
            // 获取响应结果
            IndexResponse indexResponse = getHighLevelClient().index(indexRequest,RequestOptions.DEFAULT);
            String index = indexResponse.getIndex();
            String type = indexResponse.getType();
            String ids = indexResponse.getId();
            long version = indexResponse.getVersion();

            if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                logger.info("doc indexed, index: "+ index +", type:"+ type +",id:"+ ids+",version:"+version);
             } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                logger.info("doc updated, index: "+ index +", type:"+ type +",id:"+ ids+",version:"+version);
            }
        }catch(ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                logger.error("version conflict");
            }
        }catch(Exception e){
            logger.error("execute index api failed, "+ e.toString());
        }     
    }
    
    /**
     * 根据主键id查询信息
     * @param uploadId
     */
    public static JSONObject searchPart(String uploadId) {
    	JSONObject json = new JSONObject();
    	// 查询文档
        GetRequest getRequest = new GetRequest(MemcacheUtil.getDatabaseObj("es-index-part"), MemcacheUtil.getDatabaseObj("es-type-part"),uploadId);
        try {
            // 获取响应结果
            GetResponse getResponse = getHighLevelClient().get(getRequest,RequestOptions.DEFAULT);
            String index = getResponse.getIndex();
            String type = getResponse.getType();
            String id = getResponse.getId();
            if (getResponse.isExists()) {
                long version = getResponse.getVersion();
                String sourceAsString = getResponse.getSourceAsString();
                json = JSONObject.fromObject(sourceAsString);
                logger.info("get doc, index: "+ index +", type:"+ type +",id:"+ id+",version:"+version +", source:"+ sourceAsString);
            }
        }catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                logger.warn("doc not found");
            }
        }
        catch(Exception e){
            logger.error("execute get api failed, "+ e.toString());
        }
        return json;
    }
    
    public static JSONObject criteriaQueryPart(SearchSourceBuilder sourceBuilder) {
    	JSONObject json = new JSONObject();
    	SearchRequest searchRequest = new SearchRequest(MemcacheUtil.getDatabaseObj("es-index-part"));
    	searchRequest.types(MemcacheUtil.getDatabaseObj("es-type-part"));
    	searchRequest.source(sourceBuilder);
    	SearchResponse response = null;
    	try {
    		List<String> sourceList = new ArrayList<>();
    		response = getHighLevelClient().search(searchRequest,RequestOptions.DEFAULT);
    		SearchHits hits= response.getHits();
            for (SearchHit searchHit : hits) {
                String sourceAsMap = searchHit.getSourceAsString();
                sourceList.add(sourceAsMap);
            }
            json.put("count", hits.totalHits);
            System.out.println(sourceList.toString());
            json.put("data",sourceList);
            
    	}catch(Exception e) {
    		System.out.println(e.toString());
    		logger.error("criteriaQuery failed, "+ e.toString());
    	}
    	return json;
    }

}
